package com.etc;

import org.apache.commons.io.FileUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.File;
import java.io.IOException;

/**
 * 需求：有如下数据：
 * a.txt
 * hello tom
 * hello jim
 * hello kitty
 * hello rose
 *
 * b.txt
 * hello jerry
 * hello jim
 * hello kitty
 * hello jack
 * c.txt
 * hello jerry
 * hello java
 * hello c++
 * hello c++
 *
 * 需要得到以下结果：
 * hello  a.txt-->4  b.txt-->4  c.txt-->4
 * java   c.txt-->1
 * jerry  b.txt-->1  c.txt-->1
 * ....
 */
public class JobSubitter {
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {

        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf);

        job.setJarByClass(JobSubitter.class);

        job.setMapperClass(WordStepOne.WordStepMapper.class);
        job.setReducerClass(WordStepOne.WordReduce.class);

        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        File file = new File("d:/file/output");
        if (file.exists()){
            FileUtils.deleteDirectory(file);
        }
        FileInputFormat.setInputPaths(job, new Path("d:/file/input"));
        FileOutputFormat.setOutputPath(job, new Path("d:/file/output"));

        job.setNumReduceTasks(1);

        boolean res = job.waitForCompletion(true);
        System.exit(res ? 0 : 1);

    }
}
